在上一篇,我們完成了一個簡單的「問候 Agent」,它可以接收輸入並回傳預設的回應。
這一篇,我們要讓 Agent 更進階一些,加入呼叫外部 API 來獲取即時資訊,並學習如何把它整合到 FastAPI 裡,打造一個可以即時回應天氣的「天氣 Agent」。
首先,我們需要到 氣象資料開放平臺 註冊會員。
登入後,點擊 取得授權碼,API Key 會出現在右側,請把它複製下來,等一下會用到。
接著,我們要使用的 API 是 中央氣象署開放資料平臺的資料擷取 API:
API 文件連結
這次會用到的 API 是這支:
打開後可以在這裡輸入剛剛拿到的 API Key,接著按下 Execute 查詢並查看回傳資料的結構。
查詢成功後的樣子如下:
注意:這支 API 裡有兩個參數可以根據自己的需求去選擇(這篇不會用到):locationName(縣市) 和 elementName(天氣因子),如果沒有特別選,會預設為全部資料都抓。
在上一篇我們已經有安裝過 fastapi 和 uvicorn了,而這一篇還需要用到:
pip install httpx python-dotenv
這兩個套件的用途分別是:
httpx:用來在 Python 裡發送 HTTP 請求,取得 API 回傳資料。python-dotenv:讓程式能讀 .env 檔,把裡面的內容當作「環境變數」,方便安全管理 API Key。在專案資料夾中,建立兩個檔案:WeatherAgent.py 和 .env
接著輸入以下程式碼:
.env
CWA_API_KEY=YOUR_API_KEY
將 .env 裡的 YOUR_API_KEY 替換成剛剛複製的授權碼。
WeatherAgent.py
import httpx
import os
# 讀取 .env 檔案,將環境變數載入系統
from dotenv import load_dotenv
from fastapi import FastAPI
# 載入 .env 中的環境變數
load_dotenv()
app = FastAPI(
    title="天氣 Agent API",
    description="一個簡單的 Agent,用於生成客製化天氣報告。",
    version="1.0.0",
)
@app.post("/weather")
async def get_weather(city: str) -> dict:
    """從中央氣象局 API 獲取指定城市的 36 小時天氣資訊(同步)"""
    try:
        # 中央氣象局的開放資料 API URL
        url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-C0032-001"
        # API 查詢參數
        params = {
            "Authorization": os.getenv("CWA_API_KEY"),
            "locationName": city,
            "format": "JSON"
        }
        # 使用 httpx.AsyncClient() 建立非同步連線並發送請求
        async with httpx.AsyncClient() as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            data = response.json()
        # 從回傳資料中取得所有地點資訊
        locations = data["records"]["location"]
        # 遍歷所有地點,找到與使用者輸入的城市相符的資料
        for loc in locations:
            if loc["locationName"] == city:
                # 依序取出天氣現象、降雨機率、最低溫、最高溫、體感
                wx = _get_first_param(loc, "Wx")
                pop = _get_first_param(loc, "PoP")
                min_t = _get_first_param(loc, "MinT")
                max_t = _get_first_param(loc, "MaxT")
                ci = _get_first_param(loc, "CI")
                # 整理回傳格式
                return {
                    "status": "success",
                    "report": (
                        f"{city} 天氣預報:{wx},降雨機率 {pop}%,"
                        f"氣溫 {min_t}°C ~ {max_t}°C,體感 {ci}。"
                    )
                }
        # 若找不到對應城市,回傳錯誤訊息
        return {"status": "error", "error_message": f"找不到 {city} 的天氣資料"}
    except Exception as e:
        return {"status": "error", "error_message": str(e)}
# 從指定地點的 weatherElement 中擷取第一筆資料
def _get_first_param(location: dict, element_name: str) -> str:
    """從 weatherElement 擷取指定 element 的第一筆資料"""
    for element in location["weatherElement"]:
        if element["elementName"] == element_name:
            return element["time"][0]["parameter"]["parameterName"]
    return "無資料"
這個 Agent 也有符合「感知 → 決策 → 行動」。WeatherAgent 符合「Agent 的三要素」的部分:
city。在終端執行 uvicorn WeatherAgent:app --host 127.0.0.1 --port 5678 --reload 並去到 http://localhost:5678/就會看到自己建立的 API 了。
API 展開的樣子。
輸入後點擊 Execute。
成功回傳的樣子:
這樣,我們就完成了「呼叫外部 API 並用 FastAPI 模擬 Agent」的範例。
未來想要讓 Agent 更聰明,例如加上 多步決策 或 多 Agent 協作,都可以在這個基礎上進行擴充。